/////////////////////////////////////////////////////////////////////////////
// Original code from libhdfs3. Copyright (c) 2013 - 2014, Pivotal Inc.
// All rights reserved. Author: Zhanwei Wang
/////////////////////////////////////////////////////////////////////////////
//  Modifications by Kumo Inc.
// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//



#include <kmhdfs/common/exception.h>
#include <kmhdfs/common/exception_internal.h>
#include <kmhdfs/common/session_config.h>
#include <functional>
#include <sstream>


namespace Hdfs {
    namespace Internal {
        template<typename T>
        static void CheckRangeGE(const char *key, T const &value, T const &target) {
            if (!(value >= target)) {
                std::stringstream ss;
                ss.imbue(std::locale::classic());
                ss << "Invalid configure item: \"" << key << "\", value: " << value
                        << ", expected value should be larger than " << target;
                THROW(HdfsConfigInvalid, "%s", ss.str().c_str());
            }
        }

        template<typename T>
        static void CheckMultipleOf(const char *key, const T &value, int unit) {
            if (value <= 0 || value % unit != 0) {
                THROW(HdfsConfigInvalid, "%s should be larger than 0 and be the multiple of %d.", key, unit);
            }
        }

        SessionConfig::SessionConfig(const Config &conf) {
            ConfigDefault<bool> boolValues[] = {
                {
                    &rpcTcpNoDelay, "rpc.client.connect.tcpnodelay", true
                },
                {
                    &readFromLocal, "dfs.client.read.shortcircuit", true
                },
                {
                    &addDatanode, "output.replace-datanode-on-failure", true
                },
                {
                    &notRetryAnotherNode, "input.notretry-another-node", false
                },
                {
                    &useMappedFile, "input.localread.mappedfile", true
                },
                {
                    &legacyLocalBlockReader, "dfs.client.use.legacy.blockreader.local", false
                },
                {
                    &connectToDnViaHostname, "dfs.client.use.datanode.hostname", false
                }
            };
            ConfigDefault<int32_t> i32Values[] = {
                {
                    &rpcMaxIdleTime, "rpc.client.max.idle", 10 * 1000,
                    std::bind(CheckRangeGE<int32_t>, std::placeholders::_1, std::placeholders::_2, 1)
                },
                {
                    &rpcPingTimeout, "rpc.client.ping.interval", 10 * 1000
                },
                {
                    &rpcConnectTimeout, "rpc.client.connect.timeout", 600 * 1000
                },
                {
                    &rpcReadTimeout, "rpc.client.read.timeout", 3600 * 1000
                },
                {
                    &rpcWriteTimeout, "rpc.client.write.timeout", 3600 * 1000
                },
                {
                    &rpcSocketLingerTimeout, "rpc.client.socekt.linger.timeout", -1
                },
                {
                    &rpcMaxRetryOnConnect, "rpc.client.connect.retry", 10,
                    std::bind(CheckRangeGE<int32_t>, std::placeholders::_1, std::placeholders::_2, 1)
                },
                {
                    &rpcTimeout, "rpc.client.timeout", 3600 * 1000
                },
                {
                    &defaultReplica, "dfs.default.replica", 3,
                    std::bind(CheckRangeGE<int32_t>, std::placeholders::_1, std::placeholders::_2, 1)
                },
                {
                    &inputConnTimeout, "input.connect.timeout", 600 * 1000
                },
                {
                    &inputReadTimeout, "input.read.timeout", 3600 * 1000
                },
                {
                    &inputWriteTimeout, "input.write.timeout", 3600 * 1000
                },
                {
                    &localReadBufferSize, "input.localread.default.buffersize", 1 * 1024 * 1024,
                    std::bind(CheckRangeGE<int32_t>, std::placeholders::_1, std::placeholders::_2, 1)
                },
                {
                    &prefetchSize, "dfs.prefetchsize", 10,
                    std::bind(CheckRangeGE<int32_t>, std::placeholders::_1, std::placeholders::_2, 1)
                },
                {
                    &maxGetBlockInfoRetry, "input.read.getblockinfo.retry", 3,
                    std::bind(CheckRangeGE<int32_t>, std::placeholders::_1, std::placeholders::_2, 1)
                },
                {
                    &maxLocalBlockInfoCacheSize, "input.localread.blockinfo.cachesize", 1000,
                    std::bind(CheckRangeGE<int32_t>, std::placeholders::_1, std::placeholders::_2, 1)
                },
                {
                    &maxReadBlockRetry, "input.read.max.retry", 60,
                    std::bind(CheckRangeGE<int32_t>, std::placeholders::_1, std::placeholders::_2, 1)
                },
                {
                    &chunkSize, "output.default.chunksize", 512,
                    std::bind(CheckMultipleOf<int32_t>, std::placeholders::_1, std::placeholders::_2, 512)
                },
                {
                    &packetSize, "output.default.packetsize", 64 * 1024
                },
                {
                    &blockWriteRetry, "output.default.write.retry", 10,
                    std::bind(CheckRangeGE<int32_t>, std::placeholders::_1, std::placeholders::_2, 1)
                },
                {
                    &outputConnTimeout, "output.connect.timeout", 600 * 1000
                },
                {
                    &outputReadTimeout, "output.read.timeout", 3600 * 1000
                },
                {
                    &outputWriteTimeout, "output.write.timeout", 3600 * 1000
                },
                {
                    &closeFileTimeout, "output.close.timeout", 3600 * 1000
                },
                {
                    &packetPoolSize, "output.packetpool.size", 1024
                },
                {
                    &heartBeatInterval, "output.heeartbeat.interval", 10 * 1000
                },
                {
                    &rpcMaxHARetry, "dfs.client.failover.max.attempts", 15,
                    std::bind(CheckRangeGE<int32_t>, std::placeholders::_1, std::placeholders::_2, 0)
                },
                {
                    &maxFileDescriptorCacheSize, "dfs.client.read.shortcircuit.streams.cache.size", 256,
                    std::bind(CheckRangeGE<int32_t>, std::placeholders::_1, std::placeholders::_2, 0)
                },
                {
                    &socketCacheExpiry, "dfs.client.socketcache.expiryMsec", 3000,
                    std::bind(CheckRangeGE<int32_t>, std::placeholders::_1, std::placeholders::_2, 0)
                },
                {
                    &socketCacheCapacity, "dfs.client.socketcache.capacity", 16,
                    std::bind(CheckRangeGE<int32_t>, std::placeholders::_1, std::placeholders::_2, 0)
                },
                {
                    &stripeReaderThreadPoolSize, "dfs.client.read.striped.thread-pool.size", 64
                }
            };
            ConfigDefault<int64_t> i64Values[] = {
                {
                    &defaultBlockSize, "dfs.default.blocksize", 64 * 1024 * 1024,
                    std::bind(CheckMultipleOf<int64_t>, std::placeholders::_1, std::placeholders::_2, 512)
                }
            };
            ConfigDefault<std::string> strValues[] = {
                {&defaultUri, "dfs.default.uri", "hdfs://localhost:9000"},
                {&rpcAuthMethod, "hadoop.security.authentication", "simple"},
                {&kerberosCachePath, "hadoop.security.kerberos.ticket.cache.path", ""},
                {&logSeverity, "dfs.client.log.severity", "INFO"},
                {&domainSocketPath, "dfs.domain.socket.path", ""}
            };

            for (size_t i = 0; i < TURBO_ARRAYSIZE(boolValues); ++i) {
                *boolValues[i].variable = conf.getBool(boolValues[i].key,
                                                       boolValues[i].value);

                if (boolValues[i].check) {
                    boolValues[i].check(boolValues[i].key, *boolValues[i].variable);
                }
            }

            for (size_t i = 0; i < TURBO_ARRAYSIZE(i32Values); ++i) {
                *i32Values[i].variable = conf.getInt32(i32Values[i].key,
                                                       i32Values[i].value);

                if (i32Values[i].check) {
                    i32Values[i].check(i32Values[i].key, *i32Values[i].variable);
                }
            }

            for (size_t i = 0; i < TURBO_ARRAYSIZE(i64Values); ++i) {
                *i64Values[i].variable = conf.getInt64(i64Values[i].key,
                                                       i64Values[i].value);

                if (i64Values[i].check) {
                    i64Values[i].check(i64Values[i].key, *i64Values[i].variable);
                }
            }

            for (size_t i = 0; i < TURBO_ARRAYSIZE(strValues); ++i) {
                *strValues[i].variable = conf.getString(strValues[i].key,
                                                        strValues[i].value);

                if (strValues[i].check) {
                    strValues[i].check(strValues[i].key, *strValues[i].variable);
                }
            }
        }
    }
}
